package cn.doitedu.rtmk.demo7;

import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class Demo7 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);




        /* *
         *  一、 接收用户行为事件流
         */
        // 创建kafka 行为明细topic的source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-events")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doit43-8")
                .setClientIdPrefix("doitedu-c")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");
        SingleOutputStreamOperator<UserEvent> eventStream = streamSource.map(json -> JSON.parseObject(json, UserEvent.class));

        KeyedStream<UserEvent, Long> keyedEventStream = eventStream.keyBy(UserEvent::getUser_id);


        /* *
         *  二、接收规则元数据表的cdc元数据流
         */

        MySqlSource<String> cdcSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("doit43")
                .tableList("doit43.rule_meta")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> metaCdcStream = env.fromSource(cdcSource, WatermarkStrategy.noWatermarks(), "meta_cdc");

        // 解析cdc json，封装成我们自己的RuleMeta对象
        SingleOutputStreamOperator<RuleMeta> ruleMetaStream = metaCdcStream.map(new MapFunction<String, RuleMeta>() {
            @Override
            public RuleMeta map(String cdcJson) throws Exception {

                JSONObject jsonObject = JSON.parseObject(cdcJson);
                String op = jsonObject.getString("op");

                RuleMeta ruleMeta;
                if (op.equals("d")) {
                    ruleMeta = jsonObject.getObject("before", RuleMeta.class);
                } else {
                    ruleMeta = jsonObject.getObject("after", RuleMeta.class);
                }

                // 填充op字段
                ruleMeta.setOp(op);

                // 反序列化预圈选人群的bitmap的字节数组
                byte[] rulePreSelectUsersBytes = ruleMeta.getRule_pre_select_users();
                Roaring64Bitmap preSelectUsersBitmap = Roaring64Bitmap.bitmapOf();

                if (rulePreSelectUsersBytes != null) {
                    preSelectUsersBitmap.deserialize(ByteBuffer.wrap(rulePreSelectUsersBytes));
                }
                // 将反序列化好的bitmap填充到ruleMeta对象
                ruleMeta.setPreSelectUsersBitmap(preSelectUsersBitmap);


                return ruleMeta;
            }
        });

        // 将规则元数据流广播
        MapStateDescriptor<String, RuleMeta> bcStateDesc = new MapStateDescriptor<>("meta_data", String.class, RuleMeta.class);
        BroadcastStream<RuleMeta> ruleMetaBroadcastStream = ruleMetaStream.broadcast(bcStateDesc);


        /* *
         *  三、将行为事件keyed流  连接  规则元数据广播流
         */
        keyedEventStream.connect(ruleMetaBroadcastStream)
                .process(new KeyedBroadcastProcessFunction<Long, UserEvent, RuleMeta, String>() {

                    ConcurrentHashMap<String, RuleCalculator> ruleCalculatorPool;
                    //MapState<String, RuleCalculator> ruleCalculatorPool;  // 行不通


                    GroovyClassLoader groovyClassLoader;
                    ListState<UserEvent> recentEventState;
                    MapState<String, String> oldCalculatorState;

                    String lock = "lock";

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        groovyClassLoader = new GroovyClassLoader();

                        // 用于存最近10s的用户行为事件
                        ListStateDescriptor<UserEvent> desc = new ListStateDescriptor<>("recent_events", UserEvent.class);
                        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.seconds(10)).build();
                        desc.enableTimeToLive(ttlConfig);
                        recentEventState = getRuntimeContext().getListState(desc);

                        // 用于判断规则运算机新老区别的规则id集合
                        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor<>("old_calculators", String.class, String.class);
                        oldCalculatorState = getRuntimeContext().getMapState(mapStateDescriptor);


                        ruleCalculatorPool = new ConcurrentHashMap<>();

                        // 申请一个MapState，来充当运算机池
                        //MapStateDescriptor<String, RuleCalculator> calculatorPoolDesc = new MapStateDescriptor<>("calculator_pool", String.class, RuleCalculator.class);
                        //ruleCalculatorPool = getRuntimeContext().getMapState(calculatorPoolDesc);


                    }

                    // 处理用户行为事件数据（keyed流中的数据）
                    @Override
                    public void processElement(UserEvent userEvent, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMeta, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {


                        // 制造故障
                        if(userEvent.getEvent_id().equals("ERR")  && RandomUtils.nextInt(1,10)% 3 == 0 ){
                            throw new RuntimeException("故意制造的异常.............................");
                        }


                        // 获取广播状态，用来保存元数据
                        ReadOnlyBroadcastState<String, RuleMeta> metaDataCache = ctx.getBroadcastState(new MapStateDescriptor<>("meta_data", String.class, RuleMeta.class));

                        /* *
                         * 正常的数据处理过程
                         */
                        synchronized (lock) {
                            /* *
                             * 故障后的恢复过程
                             * 判断是否是从故障中重启，从元数据缓存广播状态中，遍历每一个规则元数据，恢复出运算机
                             */
                            if (ruleCalculatorPool.isEmpty()) {
                                for (Map.Entry<String, RuleMeta> entry : metaDataCache.immutableEntries()) {
                                    String ruleId = entry.getKey();
                                    RuleMeta meta = entry.getValue();

                                    // 根据meta对象，恢复运算机，放入运算机池
                                    RuleCalculator ruleCalculator = buildRuleCalculator(meta);
                                    ruleCalculatorPool.put(ruleId, ruleCalculator);

                                    log.warn("在processElement中恢复了一个运算机,ruleId:{}",ruleId);
                                }
                            }

                            // 先把这条数据放入10s缓存
                            recentEventState.add(userEvent);

                            // 遍历运算机池
                            for (Map.Entry<String, RuleCalculator> entry : ruleCalculatorPool.entrySet()) {
                                String ruleId = entry.getKey();
                                RuleCalculator ruleCalculator = entry.getValue();
                                // 如果是一个新运算机
                                if (!oldCalculatorState.contains(ruleId)) {
                                    // 需要先把10s缓存中的历史数据，交给它处理完
                                    for (UserEvent event : recentEventState.get()) {
                                        ruleCalculator.calculate(event, out);
                                    }
                                    // 将它标记为老运算机
                                    oldCalculatorState.put(ruleId, "");
                                } else {
                                    // 如果是一个老运算机，直接把当前收到的event交给它处理
                                    ruleCalculator.calculate(userEvent, out);
                                }
                            }
                        }

                    }

                    // 处理规则元数据（broadcast流中的数据）
                    @Override
                    public void processBroadcastElement(RuleMeta ruleMeta, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMeta, String>.Context ctx, Collector<String> out) throws Exception {

                        // 获取广播状态，作为运算机池,行不通
                        //BroadcastState<String, RuleCalculator> ruleCalculatorPool = ctx.getBroadcastState(new MapStateDescriptor<String, RuleCalculator>("calculator_pool", String.class, RuleCalculator.class));


                        // 获取广播状态，用来保存元数据
                        BroadcastState<String, RuleMeta> metaDataCache = ctx.getBroadcastState(new MapStateDescriptor<>("meta_data", String.class, RuleMeta.class));


                        /**
                         * 故障后的恢复过程
                         * 判断是否是从故障中重启，从元数据缓存广播状态中，遍历每一个规则元数据，恢复出运算机
                         */
                        synchronized (lock) {
                            if (ruleCalculatorPool.isEmpty()) {
                                for (Map.Entry<String, RuleMeta> entry : metaDataCache.entries()) {
                                    String ruleId = entry.getKey();
                                    RuleMeta meta = entry.getValue();

                                    // 根据meta对象，恢复运算机，放入运算机池
                                    RuleCalculator ruleCalculator = buildRuleCalculator(meta);
                                    ruleCalculatorPool.put(ruleId, ruleCalculator);

                                    log.warn("在processBroadcastElement中恢复了一个运算机,ruleId:{}",ruleId);

                                }
                            }
                        }

                        /**
                         * 正常的规则上下线
                         */
                        String ruleId = ruleMeta.getRule_id();
                        String modelId = ruleMeta.getModel_id();
                        String op = ruleMeta.getOp();

                        if (op.equals("d")) {
                            // 根据ruleId，从规则运算机池中移除该规则的运算机对象
                            ruleCalculatorPool.remove(ruleId);

                            // 从元数据缓存中移除该规则的元数据
                            metaDataCache.remove(ruleId);

                            log.warn("下线了一个规则，规则id:{}", ruleId);
                        } else {
                            RuleCalculator ruleCalculator = buildRuleCalculator(ruleMeta);
                            // 将初始化好的规则运算机对象，放入全局运算机池
                            ruleCalculatorPool.put(ruleId, ruleCalculator);

                            // 将本次收到的规则元数据放入广播状态
                            metaDataCache.put(ruleId, ruleMeta);

                            log.warn("上线了一个规则，规则id:{}", ruleId);
                        }
                    }

                    private RuleCalculator buildRuleCalculator(RuleMeta ruleMeta) throws Exception {
                        // 用groovy的类加载工具，在运行时，根据规则附带的运算机模型代码，构造运算机实例
                        Class aClass = groovyClassLoader.parseClass(ruleMeta.getModel_calculator_source_code());
                        RuleCalculator ruleCalculator = (RuleCalculator) aClass.newInstance();

                        // 从规则元数据中，提取各种必须的信息：规则参数、规则人群、runtimeContext
                        String ruleParamJson = ruleMeta.getRule_param_json();  // 取出规则的参数
                        Roaring64Bitmap preSelectUsersBitmap = ruleMeta.getPreSelectUsersBitmap();  // 取出规则的静态预圈选人群
                        Long histQueryEndTime = ruleMeta.getHist_query_end_time();  // 取出"跨历史未来动态画像条件"的历史段预计算截止时间点

                        RuntimeContext runtimeContext = getRuntimeContext();  // 取出flink的 runtime context

                        // 利用元数据中的信息，初始化这个新的规则实例对象
                        ruleCalculator.init(runtimeContext, ruleParamJson, preSelectUsersBitmap, histQueryEndTime);

                        return ruleCalculator;

                    }
                }).print();


        env.execute();
    }


}
